package com.seari.main.dataHandler;

import com.seari.bean.RedisKey;
import com.seari.common.Const;
import com.seari.service.CommonDataService;
import com.seari.service.FirstInitService;
import com.seari.utils.CommonUtils;
import com.seari.utils.JedisPoolUtils;
import com.seari.utils.ListUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@Component
@Order(value = 2)
public class HistoryDataHandlerStarter implements ApplicationRunner {

    final static Logger logger = LoggerFactory.getLogger(HistoryDataHandlerStarter.class);

    public static final int MAX_WORKER_COUNT = 10;
    public static final int MAX_SLICE_COUNT = 20000;
    public static BlockingQueue<List<String>> aiQueue = new LinkedBlockingQueue<>(200);
    public static BlockingQueue<List<String>> diQueue = new LinkedBlockingQueue<>(200);
    public static ExecutorService service = Executors.newCachedThreadPool();
    public static AtomicInteger workerCount = new AtomicInteger();
    Jedis jedis = JedisPoolUtils.getJedis();

    List<RedisKey> aiKeyList = new ArrayList<RedisKey>();
    List<RedisKey> diKeyList = new ArrayList<RedisKey>();

    @Autowired
    private FirstInitService firstInitService;
    @Autowired
    private CommonDataService commonDataService;

    @Override
    public void run(ApplicationArguments args) throws Exception{

        initLocalCacheAndPutDataToInfluxdb();
        TimeUnit.SECONDS.sleep(20);
        logger.info("first init localCache and putData To influxdb complete");

        while (true) {

            aiKeyList = commonDataService.getRedisKeysByType(Const.REDIS_AI_KEY_TYPE);
            diKeyList = commonDataService.getRedisKeysByType(Const.REDIS_DI_KEY_TYPE);

            if(!CommonUtils.objectIsNull(aiKeyList)) {
                for(int i=0;i<aiKeyList.size();i++) {
                        String aikeyStr = aiKeyList.get(i).getKeyNameEn();
                        Set<String> aiKeyset = jedis.hkeys(aikeyStr);
                        if (!CommonUtils.objectIsNull(aiKeyset)) {
                            List<String> result = new ArrayList<String>(aiKeyset);
                            List<String> listStr = new ArrayList<String>();
                            listStr.addAll(result);
                            listStr.add(aikeyStr);
                            aiQueue.add(listStr);
                        } else {
                            logger.info("   " + aikeyStr + " key  not find in redis");
                        }
                    }
            }else{
                logger.info("   " + aiKeyList + " aiKeyList is null");
            }

            if(!CommonUtils.objectIsNull(diKeyList)) {
                for(int i=0;i<diKeyList.size();i++) {
                        String dikeyStr = diKeyList.get(i).getKeyNameEn();
                        Set<String> diKeyset = jedis.hkeys(dikeyStr);
                        if(!CommonUtils.objectIsNull(diKeyset)) {
                            List<String> result = new ArrayList<String>(diKeyset);
                            List<List<String>> listList = ListUtils.splitList(result, MAX_SLICE_COUNT);
                            for (List<String> list : listList) {
                                List<String> listStr = new ArrayList<String>();
                                listStr.addAll(list);
                                listStr.add(dikeyStr);
                                diQueue.add(listStr);
                            }
                        }else{
                           logger.info( " "+dikeyStr+ "  key not find in redis");
                        }
                    }
                }else{
                    logger.info("   " + diKeyList + " diKeyList is null");
            }

            startWorker();
            TimeUnit.SECONDS.sleep(60);
        }
    }

    public void startWorker(){

        while(workerCount.get() < MAX_WORKER_COUNT){
            HistoryDataHandlerWorker worker = new HistoryDataHandlerWorker(aiQueue,diQueue);
            service.submit(worker);
            int num = workerCount.incrementAndGet();
            logger.info("start worker:current work count is:" + num);
        }
    }

    public void initLocalCacheAndPutDataToInfluxdb(){

        firstInitService.initLocalCacheAndPutDataToInfluxdb();
    }
}
